package com.ydl.learning.flink.demo.pipeline.java.impl;

import com.ydl.learning.flink.demo.pipeline.java.Init;
import com.ydl.learning.flink.demo.pipeline.java.Process;
import com.ydl.learning.flink.demo.pipeline.java.Sink;
import com.ydl.learning.flink.demo.pipeline.java.entity.JobConfig;

/**
 * @author ydl
 * @since 2022/1/6
 */
public class Pipeline<T> {
  private JobConfig jobConfig;
  private Init init;
  private Process<T> process;
  private Sink<T> sink;

  private Pipeline(Builder<T> builder) {
    this.jobConfig = builder.jobConfig;
    this.init = builder.init;
    this.process = builder.process;
    this.sink = builder.sink;
  }

  public static <T> Builder<T> builder() {
    return new Builder<>();
  }

  public void execute() {
    //  TODO
    process.run();
  }

  public static class Builder<T> {
    private JobConfig jobConfig;
    private Init<T> init;
    private Process<T> process;
    //TODO 测流输出
    private Object processSideOut;
    //TODO 监控
    private Object processMonitor;
    private Sink<T> sink;
    //TODO 流克隆
    private Object clone;
    //TODO 克隆流处理
    private Object cloneStreamProcess;
    //TODO 克隆流输出
    private Object cloneStreamSink;

    public Builder<T> jobConfig(JobConfig jobConfig) {
      this.jobConfig = jobConfig;
      return this;
    }

    public Builder<T> init(Init init) {
      this.init = init;
      return this;
    }

    public Builder<T> process(Process<T> process) {
      this.process = process;
      return this;
    }

    public Builder<T> sink(Sink<T> sink) {
      this.sink = sink;
      return this;
    }

    public Pipeline<T> build() {
      return new Pipeline<>(this);
    }
  }
}
